package com.atguigu.gmall.realtime.dwd.app;

/*
import com.atguigu.gmall.realtime.common.base.BaseSqlApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.FlinkSqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

*/
/**
 * 交易域加购事务事实表
 *//*

public class DwdTradeCartAdd extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeCartAdd().start(10013, 4, Constant.TOPIC_DWD_TRADE_CART_ADD);
    }

    @Override
    public void handle(StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        // 读取topic_db的数据
        readOdsTopicDb(tableEnv, Constant.TOPIC_DWD_TRADE_CART_ADD);


        // 过滤出加购数据
        // database='gmall'
        // AND
        // table = 'cart_info'
        // AND
        // (
        //( type = 'insert')
        // OR (type = 'update' and old['sku_num'] is not null and data['sku_num'] > old['sku_num'])
        // )
        String filterCartAddSql =
                " select " +
                        " `data`['id'] id," +
                        " `data`['user_id'] user_id," +
                        " `data`['sku_id'] sku_id," +
                        " `data`['cart_price'] cart_price," +
                        " if(type='insert', cast(`data`['sku_num'] as int), cast(`data`['sku_num'] as int ) - cast(`old`['sku_num'] as int)) sku_num," +
                        " `data`['sku_name'] sku_name," +
                        " `data`['source_type'] source_type," +
                        " `data`['source_id'] source_id " +
                        " from topic_db " +
                        " where `database` = 'gmall' " +
                        " and `table` = 'cart_info'" +
                        " and (" +
                        " type = 'insert'" +  // 判断是否为加购数据
                        " or " +
                        " (type = 'update' and `old`['sku_num'] is not null and cast(`data`['sku_num'] as int) > cast(`old`['sku_num'] as int) )" +
                        " ) ";


        // tableEnv.sqlQuery(filterCartAddSql).execute().print();
        Table table = tableEnv.sqlQuery(filterCartAddSql);


        // 关联字典表(不需要)

        // 写出数据到Kafka
        String createSinkTable =
                " create table " + Constant.TOPIC_DWD_TRADE_CART_ADD + "(" +
                        " id STRING," +
                        " user_id STRING," +
                        " sku_id STRING, " +
                        " cart_price STRING," +
                        " sku_num INT," +
                        " sku_name STRING " +
                        " ) " + FlinkSqlUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_CART_ADD) ;

        tableEnv.executeSql(createSinkTable);

        table.executeInsert(Constant.TOPIC_DWD_TRADE_CART_ADD);

    }
}
*/

import com.atguigu.gmall.realtime.common.base.BaseSqlApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.FlinkSqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author WEIYUNHUI
 * @date 2024/8/5 9:05
 *
 * 交易域加购事务事实表
 */
public class DwdTradeCartAdd extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeCartAdd().start(10013 ,4 , Constant.TOPIC_DWD_TRADE_CART_ADD);
    }

    @Override
    public void handle(StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        // 读取topic_db的数据
        readOdsTopicDb(tableEnv , Constant.TOPIC_DWD_TRADE_CART_ADD);

        // 过滤出加购数据
        Table table = filterCartInfo(tableEnv);

        //写出到Kafka
        writerToKafka(tableEnv, table);

    }

    private static void writerToKafka(StreamTableEnvironment tableEnv, Table table) {
        String createSinkTable =
                " create table " + Constant.TOPIC_DWD_TRADE_CART_ADD + " ( " +
                        " id STRING , " +
                        " user_id STRING , " +
                        " sku_id STRING , " +
                        " cart_price STRING , " +
                        " sku_num INT , " +
                        " sku_name STRING," +
                        " ts BIGINT " +
                        " ) " + FlinkSqlUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_CART_ADD) ;

        tableEnv.executeSql(createSinkTable) ;

        table.executeInsert( Constant.TOPIC_DWD_TRADE_CART_ADD ) ;
    }

    private static Table filterCartInfo(StreamTableEnvironment tableEnv) {
        // database = 'gmall'
        // AND
        // table = 'cart_info'
        // AND
        // (
        // type = 'insert'
        // or
        // (type = 'update' and old['sku_num'] is not null  and data['sku_num'] > old['sku_num'])
        // )

        String filterCartAddSql =
                " select " +
                        "`data`['id'] id ," +
                        "`data`['user_id'] user_id," +
                        "`data`['sku_id'] sku_id," +
                        "`data`['cart_price'] cart_price," +
                        " if( type='insert' , cast(`data`['sku_num'] as int)  , cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int  ) ) sku_num," +
                        "`data`['sku_name'] sku_name ," +
                        " ts" +
                        " from topic_db " +
                        " where `database` = 'gmall' " +
                        " and `table` = 'cart_info' " +
                        " and ( " +
                        "  type = 'insert' " +
                        "  or " +
                        "  ( type ='update' and  `old`['sku_num'] is not null and cast(`data`['sku_num'] as int)>  cast(`old`['sku_num'] as int) ) " +
                        " ) ";

        Table table = tableEnv.sqlQuery(filterCartAddSql);
        return table;
    }
}